package Demo3;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;

/**
 * Created by lenovo on 2017/11/29.
 * 流式统计单词数量
 */
public class SparkStreaming_javaWC {
    public static void main(String[] args){
       SparkConf conf = new SparkConf().setAppName("SparkStreaming_javaWC").setMaster("local[2]").set("spark.testing.memory","2147480000");
      JavaStreamingContext jscc =  new JavaStreamingContext(conf, Durations.seconds(5));

       JavaReceiverInputDStream socketDStrem = jscc.socketTextStream("hadoop1",2211);

       JavaDStream<String> words = socketDStrem.flatMap(new FlatMapFunction<String,String>() {
            @Override
            public Iterable<String> call(String line) throws Exception {
                return Arrays.asList(line.split(" ")) ;
            }
        });

      JavaPairDStream<String,Integer> paris =  words.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<String, Integer>(s,1);
            }
        });

      JavaPairDStream<String,Integer> wordcounts =  paris.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1+v2;
            }
        });

        wordcounts.print();

        jscc.start();
        jscc.awaitTermination();
        jscc.close();
    }
}
